#!/usr/bin/env python


"""
Unit test for xmlsql.py

"""

__revision__ = "$Id$"

import unittest
from test import test_support
import sys
sys.path.insert(0,'../')

from xmlsql import *
from sql import *
import table
import test_main as TM
import main , main_init

XML = '<?xml version="1.0" encoding="utf-8"?>'

test = (
    '<a/>' ,
    '<a> </a>' ,
    '<a b="c"/>' ,
    '<a b="c" d="e"/>' ,
    '<a><a/></a>' ,
    '<a><a> </a></a>' ,
    '<a><a b="c"/></a>' ,
    '<a><a b="c" d="e"/></a>' ,
    '<a>abc</a>' ,
    '<a><a>abc</a></a>' ,
    '<a b="10"/>' ,
    '<a b="20" d="e"/>' ,
    '<a><a b="30"/></a>' ,
    '<a><a b="40" d="e"/></a>' ,
    '<c><a b="30"/></c>' ,
    '<c><a b="40" d="e"/></c>' ,
    '<c><ts b="40" d="e" t="2005-05-30 10:08:57"/></c>' ,
    )

DBNAME = TM.DBNAME
USER = TM.USER
INPUT_ENCODING = TM.INPUT_ENCODING

class Test_init (unittest.TestCase) :
    dsn = 'dbname='+DBNAME
    schema = 'public'
    quiet = 1
    tables = None
    xml2sqltags = None
    xml2sqlattrs = None
    sql2xmltags = None
    sql2xmlattrs = None
    def before_hook ( self ) : pass
    def setUp ( self ) :
        TM.reset_db ()
        main_init.init ( self )
        main.connect ( self )
        self.before_hook()
        self.x = XmlSql ( self.conn , getattr(self,'schema') ,
                          getattr(self,'tables') ,
                          getattr(self,'xml2sqltags') ,
                          getattr(self,'xml2sqlattrs') ,
                          getattr(self,'sql2xmltags') ,
                          getattr(self,'sql2xmlattrs') )
    def tearDown ( self ) :
        delattr ( self , 'x' )
        main.disconnect ( self )

class Test_identity (Test_init) :
    def test_identity ( self ) :
        docs = [ self.x.insert(XML+t) for t in test ]
        for d,t in zip(docs,test) :
            w = TM.nonclosableStringIO()
            self.x.extract(d,w)
            self.assertEqual(w.getvalue(),XML+t)

class Test_another_schema (Test_identity) :
    schema = 'tmp'

class Test_table (Test_identity) :
    def before_hook ( self ) :
        cr = self.conn.cursor()
        cr.execute ( """ create table a ( attr_list bigint , b text ) """ )
        self.conn.commit()
        self.tables = [ table.auto_row('a',self.conn,'public',setdefault=0) ]
    
class Test_table_int (Test_identity) :
    def before_hook ( self ) :
        cr = self.conn.cursor()
        cr.execute ( """ create table a ( attr_list bigint , b integer ) """ )
        self.conn.commit()
        self.tables = [ table.auto_row('a',self.conn,'public',setdefault=0) ]
    
class Test_table_timestamp (Test_identity) :
    def before_hook ( self ) :
        cr = self.conn.cursor()
        cr.execute ( """create table ts ( attr_list bigint , t timestamp ) """ )
        self.conn.commit()
        self.tables = [ table.auto_row('ts',self.conn,'public',setdefault=0) ]
    def test_identity ( self ) :
        t = '<ts t="2005-05-30 10:08:57"/>'
        doc =  self.x.insert(XML+t)
        w = TM.nonclosableStringIO()
        self.x.extract(doc,w)
        self.assertEqual(w.getvalue(),XML+t)
        cr = self.conn.cursor()
        cr.execute ( """select t from ts """ )
        self.assertEqual(str(cr.fetchone()[0])[:19],"2005-05-30 10:08:57")
    
class Test_table_not_exist (Test_identity) :
    def test_identity ( self ) :
        self.assertRaises ( TypeError , table.auto_row ,
                           'a',self.conn,'public',setdefault=0 )
    
class Test_xml2sqltags (Test_init) :
    xml2sqltags = lambda t : t=='a' and 'x' or t=='c' and 'y' or t
    xml2sqltags = staticmethod ( xml2sqltags )
    res = (
        '<x/>' ,
        '<x> </x>' ,
        '<x b="c"/>' ,
        '<x b="c" d="e"/>' ,
        '<x><x/></x>' ,
        '<x><x> </x></x>' ,
        '<x><x b="c"/></x>' ,
        '<x><x b="c" d="e"/></x>' ,
        '<x>abc</x>' ,
        '<x><x>abc</x></x>' ,
        '<x b="10"/>' ,
        '<x b="20" d="e"/>' ,
        '<x><x b="30"/></x>' ,
        '<x><x b="40" d="e"/></x>' ,
        '<y><x b="30"/></y>' ,
        '<y><x b="40" d="e"/></y>' ,
        )
    def test_conversion ( self ) :
        docs = [ self.x.insert(XML+t) for t in test ]
        for d,t in zip(docs,self.res) :
            w = TM.nonclosableStringIO()
            self.x.extract(d,w)
            self.assertEqual(w.getvalue(),XML+t)

class Test_xml2sqlattrs (Test_xml2sqltags) :
    xml2sqltags = None
    xml2sqlattrs = lambda t : t=='b' and 'z' or t
    xml2sqlattrs = staticmethod ( xml2sqlattrs )
    res = (
        '<a/>' ,
        '<a> </a>' ,
        '<a z="c"/>' ,
        '<a z="c" d="e"/>' ,
        '<a><a/></a>' ,
        '<a><a> </a></a>' ,
        '<a><a z="c"/></a>' ,
        '<a><a z="c" d="e"/></a>' ,
        '<a>abc</a>' ,
        '<a><a>abc</a></a>' ,
        '<a z="10"/>' ,
        '<a z="20" d="e"/>' ,
        '<a><a z="30"/></a>' ,
        '<a><a z="40" d="e"/></a>' ,
        '<c><a z="30"/></c>' ,
        '<c><a z="40" d="e"/></c>' ,
        )
class Test_sql2xmltags (Test_xml2sqltags) :
    xml2sqltags = None
    sql2xmltags = lambda t : t=='a' and 'x' or t=='c' and 'y' or t
    sql2xmltags = staticmethod ( sql2xmltags )

class Test_sql2xmlattrs (Test_xml2sqlattrs) :
    xml2sqltags = None
    xml2sqlattrs = None
    sql2xmlattrs = lambda t : t=='b' and 'z' or t
    sql2xmlattrs = staticmethod ( sql2xmlattrs )

class Test_filter (Test_init) :
    def test_filter ( self ) :
        docs = [ self.x.insert(XML+t) for t in test ]
        res = [ d for d in self.x.filter([(0,3),(2,2),(15,20)]) ]
        self.assertEqual(res,[1,2,3,2,15,16,17])

class Test_comment (Test_init) :
    def test_comment ( self ) :
        docs = [ self.x.insert(XML+t) for t in test[:3] ]
        for d,t in zip(docs,test[:3]) :
            w = TM.nonclosableStringIO()
            self.x.extract(d,w,comment=1)
            self.assertEqual(w.getvalue(),XML+'<!-- %d -->'%d+t)

class Test_table_many (Test_identity) :
    def before_hook ( self ) :
        cr = self.conn.cursor()
        cr.execute ( """
        create table a ( attr_list bigint , b integer , d text ) """ )
        cr.execute ( """
        create table c ( attr_list bigint , b integer , c text ) """ )
        self.conn.commit()
        self.tables = [ table.auto_row('a',self.conn,'public',setdefault=0) ,
                        table.auto_row('c',self.conn,'public',setdefault=0)]
    
class Test_fill (Test_init) :
    def test_fill ( self ) :
        docs = [ self.x.insert(XML+t) for t in test ]
        cr = self.conn.cursor()
        cr.execute ( """
        create table a ( attr_list bigint , b integer , d text ) """ )
        tbl = table.auto_row('a',self.conn,'public',setdefault=0)
        self.x.register_table(tbl)
        self.x.fill ( 'a' )
        cr.execute ( 'select count(*) from a' )
        self.assert_(cr.fetchone()[0]>0)
        for d,t in zip(docs,test) :
            w = TM.nonclosableStringIO()
            self.x.extract(d,w)
            self.assertEqual(w.getvalue(),XML+t)

class Test_fill_two (Test_init) :
    def test_fill ( self ) :
        docs = [ self.x.insert(XML+t) for t in test ]
        cr = self.conn.cursor()
        cr.execute ( """
        create table a ( attr_list bigint , b integer , d text ) """ )
        cr.execute ( """
        create table c ( attr_list bigint , b integer , c text ) """ )
        tbl = table.auto_row('a',self.conn,'public',setdefault=0)
        self.x.register_table(tbl)
        tbl = table.auto_row('c',self.conn,'public',setdefault=0)
        self.x.register_table(tbl)
        self.x.fill ( 'a' )
        cr.execute ( 'select count(*) from a' )
        self.assert_(cr.fetchone()[0]>0)
        self.x.fill ( 'c' )
        cr.execute ( 'select count(*) from c' )
        self.assert_(cr.fetchone()[0]>0)
        for d,t in zip(docs,test) :
            w = TM.nonclosableStringIO()
            self.x.extract(d,w)
            self.assertEqual(w.getvalue(),XML+t)

class Test_free (Test_init) :
    def before_hook ( self ) :
        cr = self.conn.cursor()
        cr.execute ( """ create table a ( attr_list bigint , b integer ) """ )
        self.conn.commit()
        self.tables = [ table.auto_row('a',self.conn,'public',setdefault=0) ]
    def test_free ( self ) :
        docs = [ self.x.insert(XML+t) for t in test ]
        self.x.free ( 'a' )
        cr = self.conn.cursor()
        cr.execute ( 'select count(*) from a' )
        self.assert_(cr.fetchone()[0]==0)
        for d,t in zip(docs,test) :
            w = TM.nonclosableStringIO()
            self.x.extract(d,w)
            self.assertEqual(w.getvalue(),XML+t)

def test_main():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(Test_init))
    suite.addTest(unittest.makeSuite(Test_identity))
    suite.addTest(unittest.makeSuite(Test_another_schema))
    suite.addTest(unittest.makeSuite(Test_table))
    suite.addTest(unittest.makeSuite(Test_table_int))
    suite.addTest(unittest.makeSuite(Test_table_timestamp))
    suite.addTest(unittest.makeSuite(Test_table_not_exist))
    suite.addTest(unittest.makeSuite(Test_xml2sqltags))
    suite.addTest(unittest.makeSuite(Test_xml2sqlattrs))
    suite.addTest(unittest.makeSuite(Test_sql2xmltags))
    suite.addTest(unittest.makeSuite(Test_sql2xmlattrs))
    suite.addTest(unittest.makeSuite(Test_filter))
    suite.addTest(unittest.makeSuite(Test_comment))
    suite.addTest(unittest.makeSuite(Test_table_many))
    suite.addTest(unittest.makeSuite(Test_fill))
    suite.addTest(unittest.makeSuite(Test_fill_two))
    suite.addTest(unittest.makeSuite(Test_free))
    test_support.run_suite(suite)

if __name__ == "__main__" :
    test_main()
#===========================================================
